//! 配置广播功能详细单元测试和Mock测试
//! 
//! 针对配置广播的每个函数进行详尽测试，覆盖所有参数组合和边界条件

#[cfg(feature = "broadcast")]
use crate::broadcast::*;
#[cfg(feature = "broadcast")]
use crate::bus_broadcast::*;
use crate::error::{ConfigResult, ConfigError};

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{SystemTime, Duration};
use tokio::sync::{RwLock, Mutex};
use async_trait::async_trait;
use serde_json::{json, Value};

// ============================================================================
// Mock 实现
// ============================================================================

/// Mock 配置广播监听器
#[derive(Debug)]
pub struct MockConfigBroadcastListener {
    name: String,
    interested_apps: Vec<String>,
    should_handle_result: bool,
    should_fail: bool,
    events_received: Arc<Mutex<Vec<ConfigBroadcastEvent>>>,
}

impl MockConfigBroadcastListener {
    pub fn new(name: String) -> Self {
        Self {
            name,
            interested_apps: vec![],
            should_handle_result: true,
            should_fail: false,
            events_received: Arc::new(Mutex::new(Vec::new())),
        }
    }

    pub fn with_interested_apps(mut self, apps: Vec<String>) -> Self {
        self.interested_apps = apps;
        self
    }

    pub fn with_should_handle(mut self, should_handle: bool) -> Self {
        self.should_handle_result = should_handle;
        self
    }

    pub fn with_failure(mut self) -> Self {
        self.should_fail = true;
        self
    }

    pub async fn get_received_events(&self) -> Vec<ConfigBroadcastEvent> {
        let events = self.events_received.lock().await;
        events.clone()
    }

    pub async fn get_events_count(&self) -> usize {
        let events = self.events_received.lock().await;
        events.len()
    }
}

#[async_trait]
impl ConfigBroadcastListener for MockConfigBroadcastListener {
    async fn on_config_change(&self, event: &ConfigBroadcastEvent) -> ConfigResult<()> {
        // 记录收到的事件
        {
            let mut events = self.events_received.lock().await;
            events.push(event.clone());
        }

        if self.should_fail {
            return Err(ConfigError::ValidationError("Mock listener failed".to_string()));
        }

        Ok(())
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn should_handle(&self, _event: &ConfigBroadcastEvent) -> bool {
        self.should_handle_result
    }

    fn interested_applications(&self) -> Vec<String> {
        self.interested_apps.clone()
    }
}

/// Mock 消息发布者
#[derive(Debug)]
pub struct MockConfigMessagePublisher {
    should_fail: bool,
    published_messages: Arc<Mutex<Vec<ConfigMessage>>>,
}

impl MockConfigMessagePublisher {
    pub fn new() -> Self {
        Self {
            should_fail: false,
            published_messages: Arc::new(Mutex::new(Vec::new())),
        }
    }

    pub fn with_failure(mut self) -> Self {
        self.should_fail = true;
        self
    }

    pub async fn get_published_messages(&self) -> Vec<ConfigMessage> {
        let messages = self.published_messages.lock().await;
        messages.clone()
    }

    pub async fn get_message_count(&self) -> usize {
        let messages = self.published_messages.lock().await;
        messages.len()
    }
}

#[async_trait]
impl ConfigMessagePublisher for MockConfigMessagePublisher {
    async fn publish(&self, message: ConfigMessage) -> Result<(), String> {
        if self.should_fail {
            return Err("Mock publisher failed".to_string());
        }

        let mut messages = self.published_messages.lock().await;
        messages.push(message);
        Ok(())
    }
}

/// Mock 消息订阅者
#[derive(Debug)]
pub struct MockConfigMessageSubscriber {
    should_fail_receive: bool,
    should_fail_subscribe: bool,
    subscribed_topics: Arc<RwLock<Vec<String>>>,
    message_queue: Arc<Mutex<Vec<ConfigMessage>>>,
    message_index: Arc<Mutex<usize>>,
}

impl MockConfigMessageSubscriber {
    pub fn new() -> Self {
        Self {
            should_fail_receive: false,
            should_fail_subscribe: false,
            subscribed_topics: Arc::new(RwLock::new(Vec::new())),
            message_queue: Arc::new(Mutex::new(Vec::new())),
            message_index: Arc::new(Mutex::new(0)),
        }
    }

    pub fn with_receive_failure(mut self) -> Self {
        self.should_fail_receive = true;
        self
    }

    pub fn with_subscribe_failure(mut self) -> Self {
        self.should_fail_subscribe = true;
        self
    }

    pub async fn add_message(&self, message: ConfigMessage) {
        let mut queue = self.message_queue.lock().await;
        queue.push(message);
    }

    pub async fn get_subscribed_topics(&self) -> Vec<String> {
        let topics = self.subscribed_topics.read().await;
        topics.clone()
    }
}

#[async_trait]
impl ConfigMessageSubscriber for MockConfigMessageSubscriber {
    async fn subscribe(&self, topic: &str) -> Result<(), String> {
        if self.should_fail_subscribe {
            return Err("Mock subscribe failed".to_string());
        }

        let mut topics = self.subscribed_topics.write().await;
        if !topics.contains(&topic.to_string()) {
            topics.push(topic.to_string());
        }
        Ok(())
    }

    async fn receive(&self) -> Result<Option<ConfigMessage>, String> {
        if self.should_fail_receive {
            return Err("Mock receive failed".to_string());
        }

        let queue = self.message_queue.lock().await;
        let mut index = self.message_index.lock().await;

        if *index < queue.len() {
            let message = queue[*index].clone();
            *index += 1;
            Ok(Some(message))
        } else {
            Ok(None)
        }
    }

    async fn unsubscribe(&self, topic: &str) -> Result<(), String> {
        let mut topics = self.subscribed_topics.write().await;
        topics.retain(|t| t != topic);
        Ok(())
    }
}

/// Mock 消息总线
#[derive(Debug)]
pub struct MockConfigMessageBus {
    should_fail_publisher: bool,
    should_fail_subscriber: bool,
    publishers: Arc<Mutex<Vec<Arc<MockConfigMessagePublisher>>>>,
    subscribers: Arc<Mutex<Vec<Arc<MockConfigMessageSubscriber>>>>,
}

impl MockConfigMessageBus {
    pub fn new() -> Self {
        Self {
            should_fail_publisher: false,
            should_fail_subscriber: false,
            publishers: Arc::new(Mutex::new(Vec::new())),
            subscribers: Arc::new(Mutex::new(Vec::new())),
        }
    }

    pub fn with_publisher_failure(mut self) -> Self {
        self.should_fail_publisher = true;
        self
    }

    pub fn with_subscriber_failure(mut self) -> Self {
        self.should_fail_subscriber = true;
        self
    }

    pub async fn get_publishers_count(&self) -> usize {
        let publishers = self.publishers.lock().await;
        publishers.len()
    }

    pub async fn get_subscribers_count(&self) -> usize {
        let subscribers = self.subscribers.lock().await;
        subscribers.len()
    }
}

#[async_trait]
impl ConfigMessageBus for MockConfigMessageBus {
    async fn create_publisher(&self) -> Result<Box<dyn ConfigMessagePublisher>, String> {
        if self.should_fail_publisher {
            return Err("Failed to create publisher".to_string());
        }

        let publisher = Arc::new(MockConfigMessagePublisher::new());
        {
            let mut publishers = self.publishers.lock().await;
            publishers.push(publisher.clone());
        }

        Ok(Box::new(MockConfigMessagePublisher::new()))
    }

    async fn create_subscriber(&self) -> Result<Box<dyn ConfigMessageSubscriber>, String> {
        if self.should_fail_subscriber {
            return Err("Failed to create subscriber".to_string());
        }

        let subscriber = Arc::new(MockConfigMessageSubscriber::new());
        {
            let mut subscribers = self.subscribers.lock().await;
            subscribers.push(subscriber.clone());
        }

        Ok(Box::new(MockConfigMessageSubscriber::new()))
    }
}

// ============================================================================
// ConfigBroadcastEventType 测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod config_broadcast_event_type_tests {
    use super::*;

    #[test]
    fn test_config_broadcast_event_type_as_str() {
        // 测试所有事件类型的字符串表示
        assert_eq!(ConfigBroadcastEventType::ConfigUpdate.as_str(), "config.update");
        assert_eq!(ConfigBroadcastEventType::ConfigDelete.as_str(), "config.delete");
        assert_eq!(ConfigBroadcastEventType::ConfigRefresh.as_str(), "config.refresh");
        assert_eq!(ConfigBroadcastEventType::ServiceRestart.as_str(), "service.restart");
    }

    #[test]
    fn test_config_broadcast_event_type_equality() {
        // 测试事件类型相等性
        assert_eq!(ConfigBroadcastEventType::ConfigUpdate, ConfigBroadcastEventType::ConfigUpdate);
        assert_ne!(ConfigBroadcastEventType::ConfigUpdate, ConfigBroadcastEventType::ConfigDelete);
    }

    #[test]
    fn test_config_broadcast_event_type_clone() {
        // 测试克隆功能
        let event_type = ConfigBroadcastEventType::ConfigUpdate;
        let cloned = event_type.clone();
        assert_eq!(event_type, cloned);
    }

    #[test]
    fn test_config_broadcast_event_type_debug() {
        // 测试Debug输出
        let event_type = ConfigBroadcastEventType::ConfigUpdate;
        let debug_str = format!("{:?}", event_type);
        assert!(debug_str.contains("ConfigUpdate"));
    }

    #[test]
    fn test_config_broadcast_event_type_serialization() {
        // 测试序列化
        let event_type = ConfigBroadcastEventType::ConfigUpdate;
        let serialized = serde_json::to_string(&event_type).unwrap();
        assert!(serialized.contains("ConfigUpdate"));
        
        // 测试反序列化
        let deserialized: ConfigBroadcastEventType = serde_json::from_str(&serialized).unwrap();
        assert_eq!(event_type, deserialized);
    }
}

// ============================================================================
// ConfigBroadcastEvent 测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod config_broadcast_event_tests {
    use super::*;

    #[test]
    fn test_config_broadcast_event_config_update_creation() {
        // 测试配置更新事件创建 - 完整参数
        let event = ConfigBroadcastEvent::config_update(
            "user-service",
            "production",
            Some("v1.0"),
            "database.url",
            Some(json!("old-url")),
            json!("new-url"),
            "config-server",
            Some("admin"),
        );

        assert_eq!(event.application, "user-service");
        assert_eq!(event.profile, "production");
        assert_eq!(event.label, Some("v1.0".to_string()));
        assert_eq!(event.config_path, Some("database.url".to_string()));
        assert_eq!(event.old_value, Some(json!("old-url")));
        assert_eq!(event.new_value, Some(json!("new-url")));
        assert_eq!(event.source_service, "config-server");
        assert_eq!(event.user, Some("admin".to_string()));
        assert_eq!(event.event_type, ConfigBroadcastEventType::ConfigUpdate);
        assert!(event.id.starts_with("config_update_"));
        assert!(event.target_services.is_none());
    }

    #[test]
    fn test_config_broadcast_event_config_update_minimal_params() {
        // 测试配置更新事件创建 - 最小参数
        let event = ConfigBroadcastEvent::config_update(
            "service",
            "dev",
            None,
            "key",
            None,
            json!("value"),
            "source",
            None,
        );

        assert_eq!(event.application, "service");
        assert_eq!(event.profile, "dev");
        assert_eq!(event.label, None);
        assert_eq!(event.user, None);
        assert_eq!(event.old_value, None);
    }

    #[test]
    fn test_config_broadcast_event_service_restart_creation() {
        // 测试服务重启事件创建 - 有目标服务
        let target_services = vec!["service1".to_string(), "service2".to_string()];
        let event = ConfigBroadcastEvent::service_restart(
            "app",
            "source-service",
            Some(target_services.clone()),
            Some("operator"),
        );

        assert_eq!(event.application, "app");
        assert_eq!(event.profile, "");
        assert_eq!(event.source_service, "source-service");
        assert_eq!(event.target_services, Some(target_services));
        assert_eq!(event.user, Some("operator".to_string()));
        assert_eq!(event.event_type, ConfigBroadcastEventType::ServiceRestart);
        assert!(event.id.starts_with("service_restart_"));
    }

    #[test]
    fn test_config_broadcast_event_service_restart_no_targets() {
        // 测试服务重启事件创建 - 无目标服务
        let event = ConfigBroadcastEvent::service_restart(
            "app",
            "source",
            None,
            None,
        );

        assert_eq!(event.target_services, None);
        assert_eq!(event.user, None);
    }

    #[test]
    fn test_config_broadcast_event_config_refresh_creation() {
        // 测试配置刷新事件创建 - 有目标服务
        let targets = vec!["target1".to_string()];
        let event = ConfigBroadcastEvent::config_refresh(
            "application",
            "staging",
            "config-server",
            Some(targets.clone()),
        );

        assert_eq!(event.application, "application");
        assert_eq!(event.profile, "staging");
        assert_eq!(event.source_service, "config-server");
        assert_eq!(event.target_services, Some(targets));
        assert_eq!(event.event_type, ConfigBroadcastEventType::ConfigRefresh);
        assert!(event.id.starts_with("config_refresh_"));
        assert_eq!(event.user, None);
    }

    #[test]
    fn test_config_broadcast_event_config_refresh_no_targets() {
        // 测试配置刷新事件创建 - 无目标服务
        let event = ConfigBroadcastEvent::config_refresh(
            "app",
            "prod",
            "source",
            None,
        );

        assert_eq!(event.target_services, None);
    }

    #[test]
    fn test_config_broadcast_event_empty_strings() {
        // 测试空字符串参数
        let event = ConfigBroadcastEvent::config_update(
            "",
            "",
            Some(""),
            "",
            None,
            json!(""),
            "",
            Some(""),
        );

        assert_eq!(event.application, "");
        assert_eq!(event.profile, "");
        assert_eq!(event.label, Some("".to_string()));
        assert_eq!(event.config_path, Some("".to_string()));
        assert_eq!(event.source_service, "");
        assert_eq!(event.user, Some("".to_string()));
    }

    #[test]
    fn test_config_broadcast_event_unicode_strings() {
        // 测试Unicode字符串
        let event = ConfigBroadcastEvent::config_update(
            "用户服务",
            "生产环境",
            Some("版本1.0"),
            "数据库.地址",
            Some(json!("旧地址")),
            json!("新地址"),
            "配置服务器",
            Some("管理员"),
        );

        assert_eq!(event.application, "用户服务");
        assert_eq!(event.profile, "生产环境");
        assert_eq!(event.label, Some("版本1.0".to_string()));
        assert_eq!(event.config_path, Some("数据库.地址".to_string()));
        assert_eq!(event.source_service, "配置服务器");
        assert_eq!(event.user, Some("管理员".to_string()));
    }

    #[test]
    fn test_config_broadcast_event_serialization() {
        // 测试事件序列化和反序列化
        let event = ConfigBroadcastEvent::config_update(
            "test-service",
            "test",
            None,
            "test.key",
            None,
            json!({"nested": "value"}),
            "source",
            None,
        );

        let serialized = serde_json::to_string(&event).unwrap();
        let deserialized: ConfigBroadcastEvent = serde_json::from_str(&serialized).unwrap();

        assert_eq!(event.application, deserialized.application);
        assert_eq!(event.profile, deserialized.profile);
        assert_eq!(event.event_type, deserialized.event_type);
    }

    #[test]
    fn test_config_broadcast_event_clone() {
        // 测试事件克隆
        let event = ConfigBroadcastEvent::config_update(
            "service",
            "profile",
            None,
            "key",
            None,
            json!("value"),
            "source",
            None,
        );

        let cloned = event.clone();
        assert_eq!(event.id, cloned.id);
        assert_eq!(event.application, cloned.application);
        assert_eq!(event.event_type, cloned.event_type);
    }
}

// ============================================================================
// ServiceState 和 BroadcastStats 测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod service_state_tests {
    use super::*;

    #[test]
    fn test_service_state_creation() {
        let mut metadata = HashMap::new();
        metadata.insert("version".to_string(), "1.0.0".to_string());
        
        let state = ServiceState {
            service_name: "test-service".to_string(),
            last_heartbeat: SystemTime::now(),
            config_version: "v1.0".to_string(),
            status: ServiceStatus::Running,
            metadata: metadata.clone(),
        };

        assert_eq!(state.service_name, "test-service");
        assert_eq!(state.config_version, "v1.0");
        assert_eq!(state.status, ServiceStatus::Running);
        assert_eq!(state.metadata, metadata);
    }

    #[test]
    fn test_service_status_equality() {
        assert_eq!(ServiceStatus::Running, ServiceStatus::Running);
        assert_ne!(ServiceStatus::Running, ServiceStatus::Stopped);
    }

    #[test]
    fn test_service_status_serialization() {
        let status = ServiceStatus::Restarting;
        let serialized = serde_json::to_string(&status).unwrap();
        let deserialized: ServiceStatus = serde_json::from_str(&serialized).unwrap();
        assert_eq!(status, deserialized);
    }

    #[test]
    fn test_broadcast_stats_default() {
        let stats = BroadcastStats::default();
        
        assert_eq!(stats.events_sent, 0);
        assert_eq!(stats.events_received, 0);
        assert_eq!(stats.events_processed, 0);
        assert_eq!(stats.events_failed, 0);
        assert_eq!(stats.active_listeners, 0);
        assert!(stats.last_processed.is_none());
    }

    #[test]
    fn test_broadcast_stats_clone() {
        let mut stats = BroadcastStats::default();
        stats.events_sent = 10;
        stats.events_received = 15;
        
        let cloned = stats.clone();
        assert_eq!(stats.events_sent, cloned.events_sent);
        assert_eq!(stats.events_received, cloned.events_received);
    }
}

// ============================================================================
// ConfigMessage 测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod config_message_tests {
    use super::*;

    #[test]
    fn test_config_message_creation() {
        let payload = json!({"test": "data"});
        let message = ConfigMessage::new(
            "test.topic".to_string(),
            payload.clone(),
        );

        assert_eq!(message.topic, "test.topic");
        assert_eq!(message.payload, payload);
        assert!(message.id.starts_with("msg_"));
        assert!(message.headers.is_empty());
        assert!(message.correlation_id.is_none());
        // 检查时间戳是最近的
        let now = SystemTime::now();
        let duration = now.duration_since(message.timestamp).unwrap_or_default();
        assert!(duration.as_secs() < 1);
    }

    #[test]
    fn test_config_message_with_header() {
        let message = ConfigMessage::new(
            "topic".to_string(),
            json!({}),
        ).with_header("key1".to_string(), "value1".to_string())
         .with_header("key2".to_string(), "value2".to_string());

        assert_eq!(message.headers.len(), 2);
        assert_eq!(message.headers.get("key1"), Some(&"value1".to_string()));
        assert_eq!(message.headers.get("key2"), Some(&"value2".to_string()));
    }

    #[test]
    fn test_config_message_with_correlation_id() {
        let correlation_id = "test-correlation-123";
        let message = ConfigMessage::new(
            "topic".to_string(),
            json!({}),
        ).with_correlation_id(correlation_id.to_string());

        assert_eq!(message.correlation_id, Some(correlation_id.to_string()));
    }

    #[test]
    fn test_config_message_chaining() {
        // 测试方法链式调用
        let message = ConfigMessage::new(
            "test.topic".to_string(),
            json!({"data": "test"}),
        )
        .with_header("type".to_string(), "config".to_string())
        .with_header("source".to_string(), "server".to_string())
        .with_correlation_id("corr-123".to_string());

        assert_eq!(message.topic, "test.topic");
        assert_eq!(message.headers.len(), 2);
        assert_eq!(message.correlation_id, Some("corr-123".to_string()));
    }

    #[test]
    fn test_config_message_empty_topic() {
        let message = ConfigMessage::new(
            "".to_string(),
            json!(null),
        );

        assert_eq!(message.topic, "");
        assert_eq!(message.payload, json!(null));
    }

    #[test]
    fn test_config_message_complex_payload() {
        let complex_payload = json!({
            "config": {
                "database": {
                    "host": "localhost",
                    "port": 5432,
                    "credentials": {
                        "username": "admin",
                        "password": "secret"
                    }
                },
                "features": ["auth", "cache", "monitoring"]
            },
            "metadata": {
                "version": "1.0.0",
                "environment": "production"
            }
        });

        let message = ConfigMessage::new(
            "config.complex".to_string(),
            complex_payload.clone(),
        );

        assert_eq!(message.payload, complex_payload);
    }

    #[test]
    fn test_config_message_unicode_content() {
        let message = ConfigMessage::new(
            "配置.主题".to_string(),
            json!({"消息": "测试数据"}),
        )
        .with_header("类型".to_string(), "配置更新".to_string())
        .with_correlation_id("关联标识-123".to_string());

        assert_eq!(message.topic, "配置.主题");
        assert_eq!(message.headers.get("类型"), Some(&"配置更新".to_string()));
        assert_eq!(message.correlation_id, Some("关联标识-123".to_string()));
    }

    #[test]
    fn test_config_message_clone() {
        let original = ConfigMessage::new(
            "topic".to_string(),
            json!({"test": "data"}),
        )
        .with_header("key".to_string(), "value".to_string())
        .with_correlation_id("corr-id".to_string());

        let cloned = original.clone();
        
        assert_eq!(original.id, cloned.id);
        assert_eq!(original.topic, cloned.topic);
        assert_eq!(original.payload, cloned.payload);
        assert_eq!(original.headers, cloned.headers);
        assert_eq!(original.correlation_id, cloned.correlation_id);
    }
}

// ============================================================================
// InMemoryConfigMessageBus 测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod in_memory_message_bus_tests {
    use super::*;

    #[tokio::test]
    async fn test_in_memory_message_bus_creation() {
        let bus = InMemoryConfigMessageBus::new();
        
        // 测试创建发布者
        let publisher_result = bus.create_publisher().await;
        assert!(publisher_result.is_ok());
        
        // 测试创建订阅者
        let subscriber_result = bus.create_subscriber().await;
        assert!(subscriber_result.is_ok());
    }

    #[tokio::test]
    async fn test_in_memory_message_bus_publish_subscribe() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        
        // 创建发布者和订阅者
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        // 订阅主题
        subscriber.subscribe("test.topic").await.unwrap();
        
        // 发布消息
        let message = ConfigMessage::new(
            "test.topic.specific".to_string(),
            json!({"data": "test"}),
        );
        publisher.publish(message.clone()).await.unwrap();
        
        // 接收消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_some());
        
        let received_msg = received.unwrap();
        assert_eq!(received_msg.topic, message.topic);
        assert_eq!(received_msg.payload, message.payload);
    }

    #[tokio::test]
    async fn test_in_memory_message_bus_multiple_messages() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        subscriber.subscribe("test").await.unwrap();
        
        // 发布多条消息
        for i in 0..5 {
            let message = ConfigMessage::new(
                format!("test.message.{}", i),
                json!({"index": i}),
            );
            publisher.publish(message).await.unwrap();
        }
        
        // 接收所有消息
        let mut received_count = 0;
        while let Ok(Some(_)) = subscriber.receive().await {
            received_count += 1;
        }
        
        assert_eq!(received_count, 5);
    }

    #[tokio::test]
    async fn test_in_memory_message_bus_topic_filtering() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        // 只订阅特定主题
        subscriber.subscribe("config.update").await.unwrap();
        
        // 发布不同主题的消息
        let config_message = ConfigMessage::new(
            "config.update.database".to_string(),
            json!({"type": "config"}),
        );
        let other_message = ConfigMessage::new(
            "system.notification".to_string(),
            json!({"type": "notification"}),
        );
        
        publisher.publish(config_message).await.unwrap();
        publisher.publish(other_message).await.unwrap();
        
        // 应该只收到匹配的消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_some());
        assert!(received.unwrap().topic.starts_with("config.update"));
        
        // 第二次调用应该没有更多消息（other_message被过滤掉了）
        let no_more = subscriber.receive().await.unwrap();
        assert!(no_more.is_none());
    }

    #[tokio::test]
    async fn test_in_memory_message_bus_unsubscribe() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        subscriber.subscribe("test.topic").await.unwrap();
        
        // 发布消息前取消订阅
        subscriber.unsubscribe("test.topic").await.unwrap();
        
        let message = ConfigMessage::new(
            "test.topic.message".to_string(),
            json!({"data": "test"}),
        );
        publisher.publish(message).await.unwrap();
        
        // 应该收不到消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_none());
    }

    #[tokio::test]
    async fn test_in_memory_message_bus_no_subscription() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        // 没有订阅任何主题
        let message = ConfigMessage::new(
            "any.topic".to_string(),
            json!({"data": "test"}),
        );
        publisher.publish(message).await.unwrap();
        
        // 应该收不到消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_none());
    }
}

// ============================================================================
// AbstractConfigBroadcastManager 测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod abstract_config_broadcast_manager_tests {
    use super::*;

    fn create_test_manager() -> AbstractConfigBroadcastManager {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        AbstractConfigBroadcastManager::new(
            bus,
            "test-service".to_string(),
            Some("test.broadcast".to_string()),
        )
    }

    #[tokio::test]
    async fn test_abstract_config_broadcast_manager_creation() {
        let manager = create_test_manager();
        
        // 测试初始状态
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 0);
        assert_eq!(stats.events_received, 0);
        assert_eq!(stats.active_listeners, 0);
        
        let cluster_state = manager.get_cluster_state().await;
        assert!(cluster_state.is_empty());
    }

    #[tokio::test]
    async fn test_broadcast_config_update() {
        let manager = create_test_manager();
        
        // 测试配置更新广播 - 完整参数
        let result = manager.broadcast_config_update(
            "user-service",
            "production",
            Some("v1.0"),
            "database.url",
            Some(json!("old-url")),
            json!("new-url"),
            Some("admin"),
        ).await;
        
        assert!(result.is_ok());
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
    }

    #[tokio::test]
    async fn test_broadcast_config_update_minimal_params() {
        let manager = create_test_manager();
        
        // 测试配置更新广播 - 最少参数
        let result = manager.broadcast_config_update(
            "service",
            "dev",
            None,
            "key",
            None,
            json!("value"),
            None,
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_broadcast_config_update_empty_strings() {
        let manager = create_test_manager();
        
        // 测试空字符串参数
        let result = manager.broadcast_config_update(
            "",
            "",
            Some(""),
            "",
            None,
            json!(""),
            Some(""),
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_broadcast_config_delete() {
        let manager = create_test_manager();
        
        // 测试配置删除广播 - 完整参数
        let result = manager.broadcast_config_delete(
            "user-service",
            "production",
            Some("v1.0"),
            "database.password",
            Some("admin"),
        ).await;
        
        assert!(result.is_ok());
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
    }

    #[tokio::test]
    async fn test_broadcast_config_delete_minimal_params() {
        let manager = create_test_manager();
        
        // 测试配置删除广播 - 最少参数
        let result = manager.broadcast_config_delete(
            "service",
            "test",
            None,
            "key",
            None,
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_broadcast_config_refresh() {
        let manager = create_test_manager();
        
        // 测试配置刷新广播 - 有目标服务
        let target_services = vec!["service1".to_string(), "service2".to_string()];
        let result = manager.broadcast_config_refresh(
            "application",
            "production",
            Some(target_services),
        ).await;
        
        assert!(result.is_ok());
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
    }

    #[tokio::test]
    async fn test_broadcast_config_refresh_no_targets() {
        let manager = create_test_manager();
        
        // 测试配置刷新广播 - 无目标服务
        let result = manager.broadcast_config_refresh(
            "app",
            "dev",
            None,
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_broadcast_service_restart() {
        let manager = create_test_manager();
        
        // 测试服务重启广播 - 有目标服务和用户
        let target_services = vec!["target-service".to_string()];
        let result = manager.broadcast_service_restart(
            "application",
            Some(target_services),
            Some("operator"),
        ).await;
        
        assert!(result.is_ok());
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
    }

    #[tokio::test]
    async fn test_broadcast_service_restart_minimal_params() {
        let manager = create_test_manager();
        
        // 测试服务重启广播 - 最少参数
        let result = manager.broadcast_service_restart(
            "app",
            None,
            None,
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_multiple_broadcasts() {
        let manager = create_test_manager();
        
        // 发送多个不同类型的广播
        manager.broadcast_config_update(
            "service1", "prod", None, "key1", None, json!("value1"), None
        ).await.unwrap();
        
        manager.broadcast_config_delete(
            "service2", "dev", None, "key2", None
        ).await.unwrap();
        
        manager.broadcast_config_refresh(
            "service3", "test", None
        ).await.unwrap();
        
        manager.broadcast_service_restart(
            "service4", None, None
        ).await.unwrap();
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 4);
    }

    #[tokio::test]
    async fn test_register_listener() {
        let manager = create_test_manager();
        
        // 注册监听器
        let listener = Box::new(MockConfigBroadcastListener::new(
            "test-listener".to_string()
        ));
        
        manager.register_listener(listener).await;
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.active_listeners, 1);
    }

    #[tokio::test]
    async fn test_register_multiple_listeners() {
        let manager = create_test_manager();
        
        // 注册多个监听器
        for i in 0..3 {
            let listener = Box::new(MockConfigBroadcastListener::new(
                format!("listener-{}", i)
            ));
            manager.register_listener(listener).await;
        }
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.active_listeners, 3);
    }

    #[tokio::test]
    async fn test_remove_listener() {
        let manager = create_test_manager();
        
        // 注册监听器
        let listener = Box::new(MockConfigBroadcastListener::new(
            "test-listener".to_string()
        ));
        manager.register_listener(listener).await;
        
        // 移除监听器
        let removed = manager.remove_listener("test-listener").await;
        assert!(removed);
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.active_listeners, 0);
    }

    #[tokio::test]
    async fn test_remove_nonexistent_listener() {
        let manager = create_test_manager();
        
        // 尝试移除不存在的监听器
        let removed = manager.remove_listener("nonexistent").await;
        assert!(!removed);
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.active_listeners, 0);
    }

    #[tokio::test]
    async fn test_manager_start_success() {
        let manager = create_test_manager();
        
        // 测试启动管理器
        let result = manager.start().await;
        assert!(result.is_ok());
        
        // 检查运行状态
        let running = *manager.running.read().await;
        assert!(running);
    }

    #[tokio::test]
    async fn test_manager_start_already_running() {
        let manager = create_test_manager();
        
        // 第一次启动应该成功
        manager.start().await.unwrap();
        
        // 第二次启动应该失败
        let result = manager.start().await;
        assert!(result.is_err());
        
        if let Err(ConfigError::ValidationError(msg)) = result {
            assert!(msg.contains("已在运行"));
        } else {
            panic!("Expected ValidationError");
        }
    }

    #[tokio::test]
    async fn test_manager_stop_success() {
        let manager = create_test_manager();
        
        // 启动后停止
        manager.start().await.unwrap();
        let result = manager.stop().await;
        assert!(result.is_ok());
        
        // 检查运行状态
        let running = *manager.running.read().await;
        assert!(!running);
    }

    #[tokio::test]
    async fn test_manager_stop_not_running() {
        let manager = create_test_manager();
        
        // 未启动就停止应该成功（幂等操作）
        let result = manager.stop().await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_manager_lifecycle() {
        let manager = create_test_manager();
        
        // 测试完整生命周期
        assert!(manager.start().await.is_ok());
        
        // 在运行状态下进行操作
        manager.broadcast_config_update(
            "test", "prod", None, "key", None, json!("value"), None
        ).await.unwrap();
        
        assert!(manager.stop().await.is_ok());
        
        // 停止后可以重新启动
        assert!(manager.start().await.is_ok());
        assert!(manager.stop().await.is_ok());
    }

    #[tokio::test]
    async fn test_unicode_parameters() {
        let manager = create_test_manager();
        
        // 测试Unicode参数
        let result = manager.broadcast_config_update(
            "用户服务",
            "生产环境",
            Some("版本1.0"),
            "数据库.配置.地址",
            Some(json!("旧地址")),
            json!("新地址"),
            Some("管理员"),
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_large_payload() {
        let manager = create_test_manager();
        
        // 创建大型payload
        let mut large_config = serde_json::Map::new();
        for i in 0..1000 {
            large_config.insert(
                format!("config_key_{}", i),
                json!(format!("config_value_{}_with_some_longer_content_to_make_it_bigger", i))
            );
        }
        
        let result = manager.broadcast_config_update(
            "service",
            "prod",
            None,
            "large.config",
            None,
            json!(large_config),
            None,
        ).await;
        
        assert!(result.is_ok());
    }
}

// ============================================================================
// Mock 组件行为测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod mock_component_tests {
    use super::*;

    #[tokio::test]
    async fn test_mock_listener_basic_functionality() {
        let listener = MockConfigBroadcastListener::new("test".to_string());
        
        assert_eq!(listener.name(), "test");
        assert_eq!(listener.interested_applications(), Vec::<String>::new());
        assert!(listener.should_handle(&ConfigBroadcastEvent::config_update(
            "any", "any", None, "any", None, json!("any"), "any", None
        )));
        
        let events_count = listener.get_events_count().await;
        assert_eq!(events_count, 0);
    }

    #[tokio::test]
    async fn test_mock_listener_with_interested_apps() {
        let listener = MockConfigBroadcastListener::new("test".to_string())
            .with_interested_apps(vec!["app1".to_string(), "app2".to_string()]);
        
        let apps = listener.interested_applications();
        assert_eq!(apps.len(), 2);
        assert!(apps.contains(&"app1".to_string()));
        assert!(apps.contains(&"app2".to_string()));
    }

    #[tokio::test]
    async fn test_mock_listener_should_handle_config() {
        let listener = MockConfigBroadcastListener::new("test".to_string())
            .with_should_handle(false);
        
        let event = ConfigBroadcastEvent::config_update(
            "any", "any", None, "any", None, json!("any"), "any", None
        );
        
        assert!(!listener.should_handle(&event));
    }

    #[tokio::test]
    async fn test_mock_listener_event_processing() {
        let listener = MockConfigBroadcastListener::new("test".to_string());
        
        let event = ConfigBroadcastEvent::config_update(
            "service", "prod", None, "key", None, json!("value"), "source", None
        );
        
        // 处理事件
        let result = listener.on_config_change(&event).await;
        assert!(result.is_ok());
        
        // 检查事件是否被记录
        let events = listener.get_received_events().await;
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].application, "service");
    }

    #[tokio::test]
    async fn test_mock_listener_failure() {
        let listener = MockConfigBroadcastListener::new("test".to_string())
            .with_failure();
        
        let event = ConfigBroadcastEvent::config_update(
            "service", "prod", None, "key", None, json!("value"), "source", None
        );
        
        let result = listener.on_config_change(&event).await;
        assert!(result.is_err());
        
        if let Err(ConfigError::ValidationError(msg)) = result {
            assert_eq!(msg, "Mock listener failed");
        } else {
            panic!("Expected ValidationError");
        }
    }

    #[tokio::test]
    async fn test_mock_publisher_basic_functionality() {
        let publisher = MockConfigMessagePublisher::new();
        
        let message = ConfigMessage::new(
            "test.topic".to_string(),
            json!({"test": "data"}),
        );
        
        let result = publisher.publish(message.clone()).await;
        assert!(result.is_ok());
        
        let published = publisher.get_published_messages().await;
        assert_eq!(published.len(), 1);
        assert_eq!(published[0].topic, message.topic);
    }

    #[tokio::test]
    async fn test_mock_publisher_failure() {
        let publisher = MockConfigMessagePublisher::new().with_failure();
        
        let message = ConfigMessage::new(
            "test.topic".to_string(),
            json!({"test": "data"}),
        );
        
        let result = publisher.publish(message).await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), "Mock publisher failed");
    }

    #[tokio::test]
    async fn test_mock_subscriber_basic_functionality() {
        let subscriber = MockConfigMessageSubscriber::new();
        
        // 订阅
        let result = subscriber.subscribe("test.topic").await;
        assert!(result.is_ok());
        
        let topics = subscriber.get_subscribed_topics().await;
        assert_eq!(topics.len(), 1);
        assert_eq!(topics[0], "test.topic");
        
        // 接收（应该没有消息）
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_none());
    }

    #[tokio::test]
    async fn test_mock_subscriber_with_messages() {
        let subscriber = MockConfigMessageSubscriber::new();
        
        // 添加消息
        let message = ConfigMessage::new(
            "test.topic".to_string(),
            json!({"test": "data"}),
        );
        subscriber.add_message(message.clone()).await;
        
        // 接收消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_some());
        
        let received_msg = received.unwrap();
        assert_eq!(received_msg.topic, message.topic);
    }

    #[tokio::test]
    async fn test_mock_subscriber_failures() {
        let subscriber = MockConfigMessageSubscriber::new()
            .with_subscribe_failure()
            .with_receive_failure();
        
        // 订阅失败
        let result = subscriber.subscribe("test.topic").await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), "Mock subscribe failed");
        
        // 接收失败
        let result = subscriber.receive().await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), "Mock receive failed");
    }

    #[tokio::test]
    async fn test_mock_bus_basic_functionality() {
        let bus = MockConfigMessageBus::new();
        
        // 创建发布者
        let publisher_result = bus.create_publisher().await;
        assert!(publisher_result.is_ok());
        
        // 创建订阅者
        let subscriber_result = bus.create_subscriber().await;
        assert!(subscriber_result.is_ok());
        
        // 检查计数
        assert_eq!(bus.get_publishers_count().await, 1);
        assert_eq!(bus.get_subscribers_count().await, 1);
    }

    #[tokio::test]
    async fn test_mock_bus_failures() {
        let bus = MockConfigMessageBus::new()
            .with_publisher_failure()
            .with_subscriber_failure();
        
        // 创建发布者失败
        let publisher_result = bus.create_publisher().await;
        assert!(publisher_result.is_err());
        let error_msg = publisher_result.unwrap_err();
        assert_eq!(error_msg, "Failed to create publisher");
        
        // 创建订阅者失败
        let subscriber_result = bus.create_subscriber().await;
        assert!(subscriber_result.is_err());
        let error_msg = subscriber_result.unwrap_err();
        assert_eq!(error_msg, "Failed to create subscriber");
    }

    #[tokio::test]
    async fn test_mock_bus_multiple_components() {
        let bus = MockConfigMessageBus::new();
        
        // 创建多个发布者和订阅者
        for _ in 0..3 {
            let _ = bus.create_publisher().await.unwrap();
            let _ = bus.create_subscriber().await.unwrap();
        }
        
        assert_eq!(bus.get_publishers_count().await, 3);
        assert_eq!(bus.get_subscribers_count().await, 3);
    }
}

// ============================================================================
// 边界条件和错误处理测试
// ============================================================================

#[cfg(test)]
#[cfg(feature = "broadcast")]
mod boundary_and_error_tests {
    use super::*;

    #[tokio::test]
    async fn test_manager_with_failed_bus() {
        let bus = Arc::new(MockConfigMessageBus::new().with_publisher_failure());
        let manager = AbstractConfigBroadcastManager::new(
            bus,
            "test-service".to_string(),
            None,
        );
        
        // 尝试发送事件应该失败
        let result = manager.broadcast_config_update(
            "service", "prod", None, "key", None, json!("value"), None
        ).await;
        
        assert!(result.is_err());
        
        if let Err(ConfigError::NetworkError(msg)) = result {
            assert!(msg.contains("创建发布者失败"));
        } else {
            panic!("Expected NetworkError");
        }
    }

    #[tokio::test]
    async fn test_extremely_long_strings() {
        let manager = create_test_manager();
        
        // 创建极长的字符串
        let long_string = "a".repeat(10000);
        
        let result = manager.broadcast_config_update(
            &long_string,
            &long_string,
            Some(&long_string),
            &long_string,
            None,
            json!(long_string.clone()),
            Some(&long_string),
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_special_characters() {
        let manager = create_test_manager();
        
        let special_chars = "!@#$%^&*()_+-=[]{}|;':,.<>?/~`";
        
        let result = manager.broadcast_config_update(
            special_chars,
            special_chars,
            Some(special_chars),
            special_chars,
            None,
            json!(special_chars),
            Some(special_chars),
        ).await;
        
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_null_and_empty_json_values() {
        let manager = create_test_manager();
        
        // 测试null值
        let result1 = manager.broadcast_config_update(
            "service", "prod", None, "key", None, json!(null), None
        ).await;
        assert!(result1.is_ok());
        
        // 测试空对象
        let result2 = manager.broadcast_config_update(
            "service", "prod", None, "key", None, json!({}), None
        ).await;
        assert!(result2.is_ok());
        
        // 测试空数组
        let result3 = manager.broadcast_config_update(
            "service", "prod", None, "key", None, json!([]), None
        ).await;
        assert!(result3.is_ok());
    }

    #[tokio::test]
    async fn test_concurrent_operations() {
        let manager = Arc::new(create_test_manager());
        
        // 并发执行多个操作
        let mut handles = vec![];
        
        for i in 0..10 {
            let manager_clone = manager.clone();
            let handle = tokio::spawn(async move {
                manager_clone.broadcast_config_update(
                    &format!("service-{}", i),
                    "prod",
                    None,
                    "key",
                    None,
                    json!(format!("value-{}", i)),
                    None,
                ).await
            });
            handles.push(handle);
        }
        
        // 等待所有操作完成
        for handle in handles {
            let result = handle.await.unwrap();
            assert!(result.is_ok());
        }
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 10);
    }

    #[tokio::test]
    async fn test_listener_registration_race_condition() {
        let manager = Arc::new(create_test_manager());
        
        // 并发注册监听器
        let mut handles = vec![];
        
        for i in 0..5 {
            let manager_clone = manager.clone();
            let handle = tokio::spawn(async move {
                let listener = Box::new(MockConfigBroadcastListener::new(
                    format!("listener-{}", i)
                ));
                manager_clone.register_listener(listener).await;
            });
            handles.push(handle);
        }
        
        // 等待所有注册完成
        for handle in handles {
            handle.await.unwrap();
        }
        
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.active_listeners, 5);
    }

    fn create_test_manager() -> AbstractConfigBroadcastManager {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        AbstractConfigBroadcastManager::new(
            bus,
            "test-service".to_string(),
            Some("test.broadcast".to_string()),
        )
    }
}